package com.snopy.mq.kafka.serial;

import com.snopy.pojo.Banana;
import org.apache.kafka.common.header.Headers;
import org.apache.kafka.common.serialization.DoubleSerializer;
import org.apache.kafka.common.serialization.Serializer;

import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.util.Arrays;
import java.util.Map;

/**
 * @author snopy
 * @version 1.0.0
 * @classname BananaSerializer
 * @description
 * @email 77912204@qq.com
 * @date 2022/06/09 22:30
 */
public class BananaSerializer implements Serializer<Banana> {

    /**
     * Convert {@code data} into a byte array.
     *
     * @param topic topic associated with data
     * @param data  typed data
     * @return serialized bytes
     */
    @Override
    public byte[] serialize(String topic, Banana data) {
        int len = 0;
        if (data==null){
            return new byte[0];
        }
        byte[] colorBytes = null;
        if (null != data.getColor()){
            colorBytes = data.getColor().getBytes(StandardCharsets.UTF_8);
        }else {
            colorBytes = new byte[0];
        }
        byte[] nameBytes = null;
        if (null != data.getName()){
            nameBytes = data.getName().getBytes(StandardCharsets.UTF_8);
        }else {
           nameBytes = new byte[0];
        }
        byte[] weightBytes = null;
        if (null != data.getWeight()){
            DoubleSerializer doubleSerializer = new DoubleSerializer();
            weightBytes =  doubleSerializer.serialize(topic, data.getWeight());
        }else {
            weightBytes = new byte[0];
        }
        byte[] descBytes = null;
        if (null != data.getDesc()){
            descBytes = data.getDesc().getBytes(StandardCharsets.UTF_8);
        }else {
            descBytes = new byte[0];
        }
        ByteBuffer buffer = ByteBuffer.allocate(16+colorBytes.length+nameBytes.length+weightBytes.length+descBytes.length);
        buffer.putInt(colorBytes.length);
        buffer.put(colorBytes);
        buffer.putInt(nameBytes.length);
        buffer.put(nameBytes);
        buffer.putInt(weightBytes.length);
        buffer.put(weightBytes);
        buffer.putInt(descBytes.length);
        buffer.put(descBytes);
        return buffer.array();
    }

    public static void main(String[] args) {
        Banana banana = new Banana(null,"snopy",11,"test");
        byte[] tests = new BananaSerializer().serialize("test", banana);
        System.out.println(Arrays.toString(tests));

        Banana banana1 = new BananaDeserializer().deserialize("test", tests);
        System.out.println(banana1);
    }
}
